package com.alison.datastream.chapter3_sink;

import com.alison.tableapisql.chapter1_tableapiandsql.model.SensorReading;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.time.Instant;

/**
 * @Author alison
 * @Date 2024/4/9 0:13
 * @Version 1.0
 * @Description
 */
public class S5_SinkTest4_Jdbc {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        String inputPath = "D:/workspace/lab/learnbigdata/learnflink/flink-datastream/src/main/resources/datastream/sensor.txt";

        env.readTextFile(inputPath)
                .map(line -> {
                    String[] fields = line.split(",");
                    return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
                })
                .addSink(JdbcSink.sink(
                        "insert into sensor (id, ts, temperature) values (?,?,?)",
                        new JdbcStatementBuilder<SensorReading>() {
                            @Override
                            public void accept(PreparedStatement ps, SensorReading t) throws SQLException {
                                ps.setString(1, t.id);
                                ps.setTimestamp(2, Timestamp.from(Instant.ofEpochSecond(t.timestamp)));
                                ps.setDouble(3, t.temperature);
                            }
                        },
                        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                                .withUrl("jdbc:mysql://localhost:3306/flink?useUnicode=true&serverTimezone=Asia/Shanghai&characterEncoding=UTF-8&useSSL=false")
                                .withDriverName("com.mysql.cj.jdbc.Driver")
                                .withUsername("root")
                                .withPassword("root")
                                .build()));
        env.execute();
    }
}
